package top.doe.flinksql.demos;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/10/23
 * @Desc: 学大数据，上多易教育
 *  反面示例： tumble窗口聚合，不支持 changelog流（update/delete）
 **/
public class Demo10_Sql_Test {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        //tenv.getConfig().set("table.exec.source.idle-timeout","10 ms");


        // 建表，映射 mysql的 cdc
        tenv.executeSql(
                " create table t1 (                                "+
                        "     id int primary key not enforced,           "+
                        "     name string,                               "+
                        "     gender string,                             "+
                        "     age int,                                   "+
                        "     score int,                                 "+
                        "     ts bigint,                                 "+
                        "     rt as to_timestamp_ltz(ts,3),              "+
                        "     watermark for rt as rt                     "+
                        " ) with (                             "+
                        "     'connector' = 'mysql-cdc',       "+
                        "     'hostname' = 'doitedu',          "+
                        "     'port' = '3306',                 "+
                        "     'username' = 'root',             "+
                        "     'password' = 'root',    "+
                        "     'database-name' = 'doit50',      "+
                        "     'table-name' = 't_test'          "+
                        " )                                    "
        );

        // 查询sql
        // 会直接报错：
        // StreamPhysicalWindowAggregate doesn't support consuming update and delete changes
        // which is produced by node TableSourceScan(table=[[default_catalog, default_database, order_cdc]], fields=[oid, uid, amt, create_time, status])
        tenv.executeSql(
                " select\n" +
                        "    tumble_start(rt,interval '5' second) as window_start, \n" +
                        "    tumble_end(rt,interval '5' second) as window_end,\n" +
                        "    gender,\n" +
                        "    sum(score) as sum_score,\n" +
                        "    count(1) as cnt,\n" +
                        "    avg(score) as avg_score\n" +
                        "from t1\n" +
                        "group by tumble(rt,interval '5' second),gender\n"
        ).print();



    }
}
